package com.ht.api.db.hlink.function.impl;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import com.ht.api.db.Db;
import com.ht.api.db.DbConfig;
import com.ht.api.db.Entity;
import com.ht.api.db.StreamLoadConfig;
import com.ht.api.db.bean.Column;
import com.ht.api.db.hlink.function.HlinkFunction;
import com.ht.api.db.task.HtTaskExecutor;
import com.ht.api.db.util.HtUtil;
import com.ht.api.db.util.RedisUtil;
import com.ht.api.db.util.SysParamUtil;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import lombok.extern.slf4j.Slf4j;

/**
 * 根据user_main表的users字段来计算oneId
 * @author asq
 * @createTime 2025年1月19日 10:02:25
 */
@Slf4j
public class OneIdFunction implements HlinkFunction<String>, AutoCloseable {
	/** redis存储oneid的分片数量（每个租户） */
    private static final int HASH_SLOTS = 100;
    /** redis中保存oneid的hash键名前缀 */
    private static final String KEY = "oneid:mapping:";
	/** 保存要批量导入数据库的各种entity，攒批处理 */
	private static List<Entity> beanList = new CopyOnWriteArrayList<>();
	/** 锁对象 */
	private static final Object LOCK = new Object();
	/** 多久保存一次数据库（秒） */
	private static final long TIME = 60;
	
    // 定时任务执行器
    private static ScheduledExecutorService scheduler;
    private static volatile boolean isRunning = false;
	
	@Override
	public String format(DataSource ds, Object value, Entity entity) {
		// 如果已经有oneid了，则跳过
		if (entity != null && StrUtil.isNotBlank(entity.getStr("oneid"))) {
			return entity.getStr("oneid");
		}
		
		// 开启定时保存任务
		startAutoSaveTask(ds);
        
        // 根据系统配置，使用不同的获取方式
        long oneId = 0;
        try {
			oneId = getOneId(ds, getUserProfileTables(), entity);
        } catch (Exception e) {
        	log.error("[oneid]获取oneid发生异常：{}", e.getMessage(), e);
        }
		
		return String.valueOf(oneId);
	}

    /**
     * 获取所有的用户画像表
     * @return
     */
    public List<String> getUserProfileTables(){
    	String str = SysParamUtil.getStr("user_profile_tables");
    	if (StrUtil.isNotBlank(str)) {
    		return Arrays.asList(str.split(","));
    	} else {
    		return new ArrayList<>();
    	}
    }
    
    @Override
    public void close() {
        if (scheduler != null) {
            scheduler.shutdown();
            try {
                if (!scheduler.awaitTermination(1, TimeUnit.MINUTES)) {
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e) {
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
            isRunning = false;
        }
    }
	
	/**
	 * 批量模式：从redis中获取oneid<br>
	 * 从redis中匹配合适的oneid，然后创建user_main数据，放入userMainMap集合，由定时任务每分钟用StreamLoad的方式导入数据库
	 * @param ds jdbc数据源
	 * @param userProfileTables 所有的用户画像表
	 * @param entity
	 * @return
	 */
	public Long getOneId(DataSource ds, List<String> userProfileTables, JSONObject entity) {
		// 将所有需要用来匹配的字段组合成一个key列表，拿去redis查询oneid
		String unionid = entity.getStr("unionid");
		String openid = entity.getStr("openid");
		String userid = entity.getStr("userid");
		List<String> values = Arrays.asList(unionid, openid, userid);
		List<String> fields = values.stream().filter(f -> StrUtil.isNotBlank(f)).toList();
		if (fields.isEmpty()) return null;
		String hashCodes = fields.stream().map(s -> String.valueOf(Math.abs(s.hashCode()))).collect(Collectors.joining(","));
		
		// 获取租户ID
		long tenantId = DbConfig.getTenantId(ds);
		
		// 从redis查找oneid
		List<Long> oneidListAll = mgetFromRedis(tenantId, fields);
		List<Long> oneidList = oneidListAll.stream().distinct().filter(oneid -> oneid != null).toList();
		
		// 如果没找到
		long oneId = 0;
		if (oneidList == null || oneidList.isEmpty()) {
			oneId = HtUtil.getIdLong();
			
			// 有新oneid时，需要往26张表中写入数据：user_main一条，25张用户画像表各一条
			addToMap(createEntityForInsert(userProfileTables, entity, oneId, hashCodes));
			
			// 将所有的ID字段以及oneid，作为key分片写入redis的hash字段中
			Map<String, Object> map = new HashMap<>();
			for (String field : fields) {
				map.put(field, oneId);
			}
			map.put(String.valueOf(oneId), new ArrayList<>(fields));
			msetToRedis(tenantId, map);
		} else {
			// 如果找到了多个，就命中第一个oneid
			oneId = oneidList.get(0);
			
			// 更新redis，得到oneid对应的字段值列表
			Set<String> allValueList = updateRedis(tenantId, oneidList, oneId, fields);
			
			// 更新数据库user_main表
			String allHashCodes = allValueList.stream()
					.map(s -> String.valueOf(Math.abs(s.hashCode())))
					.collect(Collectors.joining(","));
			addToMap(createEntityForInsert(userProfileTables, entity, oneId, allHashCodes));
			
			// 如果找到的不止1个，还需要处理数据库
			if (oneidList.size() > 1) {
				// 将数据库中不用的oneid数据，作个合并标记
				for (int i = 1; i < oneidList.size(); i++) {
					long moreOneId = oneidList.get(i);
					addToMap(createEntityForMerge(ds, userProfileTables, oneId, moreOneId));
				}
			}
		}
		
		return oneId;
	}

	/**
	 * 匹配到多个oneid时，更新redis
     * @param tenantId 租户ID
	 * @param oneidList 匹配到的oneid列表
	 * @param oneId 最终选用的oneId
	 * @param fields 本次数据的对应字段列表
	 * @return oneid对应的所有unionid,userid等
	 */
	private Set<String> updateRedis(long tenantId, List<Long> oneidList, final long oneId, List<String> fields) {
		// 取出所有oneid的key，一次性查询出这些oneid的value
		List<String> moreOneIdkeys = oneidList.stream().map(oneid -> String.valueOf(oneid)).toList();
		List<List<String>> moreOneIdValues = mgetFromRedis(tenantId, moreOneIdkeys);

		// 将这些value组装成一个大集合，再重新写入被命中的那个oneid上
		Set<String> allValueList = new LinkedHashSet<>();
		moreOneIdValues.stream().filter(c -> c != null).forEach(valueList -> {
			allValueList.addAll(valueList);
		});
		
		// 再加上本次传递过来的所有字段的hashcode
		allValueList.addAll(fields);
		
		// 写入redis
		Map<String, Object> map = new HashMap<>();
		allValueList.stream().forEach(field -> map.put(field, oneId));
		map.put(String.valueOf(oneId), new ArrayList<>(allValueList));
		msetToRedis(tenantId, map);

		return allValueList;
	}
	
	/**
	 * 命中oneid时，创建需要新增或修改数据的entity对象<br>
	 * 这个列表包含一个user_main对象以及所有用户画像表的entity对象
	 * @param userProfileTables 所有的用户画像表
	 * @param entity
	 * @param oneId
	 * @param users user_main的users字段值
	 * @return
	 */
	private List<Entity> createEntityForInsert(List<String> userProfileTables, JSONObject entity, long oneId, String users) {
		List<Entity> list = new ArrayList<>();
		
		// 添加user_main对象
		Entity userMainEntity = Entity.create("user_main")
				.set("main_id", oneId)
				.set("oneid", oneId)
				.set("nickname", entity.getStr("nickname"))
				.set("avatar", entity.getStr("avatar"))
				.set("aliasname", "")
				.set("gender", entity.getInt("gender"))
				.set("status", 0)
				.set("first_in_time", entity.getStr("in_time"))
				.set("last_exit_time", entity.getStr("exit_time"))
				.set("merge_oneid", 0);
		if (StrUtil.isNotBlank(users)) {
			userMainEntity.set("users", users);
		}
		list.add(userMainEntity);
		
		// 添加所有用户画像表对象
		userProfileTables.stream().forEach(t -> {
			Entity userProfileEntity = Entity.create(t).set("main_id", oneId).set("oneid", oneId);
			setNullTo0(userProfileEntity);
			list.add(userProfileEntity);
		});
		
		return list;
	}
	
	/**
	 * 将字段类型为int/decimal并且函数为sum的，设置默认值为0
	 * @param entity
	 */
	private void setNullTo0(Entity entity) {
		String table = entity.getTable();
		List<Column> columnList = DbConfig.columnList(table);
		columnList.stream().forEach(c -> {
			if (("decimal".equalsIgnoreCase(c.getType()) || "int".equalsIgnoreCase(c.getType())) 
					&& "SUM".equalsIgnoreCase(c.getFun())) {
				entity.set(c.getCode(), 0);
			}
		});
	}
	
	/**
	 * 合并oneid时，创建需要修改数据的entity对象<br>
	 * 这个列表包含一个user_main对象以及所有用户画像表的entity对象
	 * @param userProfileTables 所有的用户画像表
	 * @param oneId 命中的oneid
	 * @param moreOneId 被合并的多余oneid
	 * @return
	 */
	private List<Entity> createEntityForMerge(DataSource ds, List<String> userProfileTables, long oneId, long moreOneId) {
		List<Entity> list = new ArrayList<>();
		
		// 修改user_main对象
		list.add(Entity.create("user_main").set("main_id", moreOneId).set("status", 1).set("merge_oneid", oneId));
		
		// 修改所有用户画像表对象
		userProfileTables.stream().forEach(t -> {
			Entity userProfile = Db.createSqlParser(t).findById(ds, moreOneId);
			if (userProfile != null) {
				userProfile.set("main_id", oneId);
				userProfile.set("oneid", oneId);
				list.add(userProfile);
			}
		});
		
		return list;
	}
	
	/**
	 * 创建每分钟写数据库的定时任务
	 * @param ds
	 */
    private synchronized void startAutoSaveTask(DataSource ds) {
        if (!isRunning) {
        	if (scheduler == null) {
	            scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
	                Thread thread = new Thread(r, "OneId-Saver");
	                thread.setDaemon(true);  // 设置为守护线程
	                return thread;
	            });
        	}
            
            // 1分钟后开始自动保存任务
            scheduler.schedule(() -> saveToDb(ds), 1, TimeUnit.SECONDS);
            
            // 标记为已开启定时任务
            isRunning = true;
        }
    }
    
    /**
     * 使用StreamLoad的方式将数据导入数据库
     * @param ds jdbc数据源（会根据此数据源，找到StreamLoad配置）
     */
    private void saveToDb(DataSource ds) {
    	// 一直循环
    	while(true) {
	        synchronized (LOCK) {
	            if (beanList.isEmpty()) {
	                try {
	                    LOCK.wait();  // 如果map为空，就等待
	                } catch (InterruptedException e) {
	                    Thread.currentThread().interrupt();
	                    return;
	                }
	            }
	            
	            // 导入数据库
	            batchStreamLoad(ds);
	        }
	            
            // 休眠1分钟
            try {
				Thread.sleep(1000 * TIME);
			} catch (InterruptedException e) {
				log.error("线程休眠发生异常：{}", e.getMessage(), e);
			}
    	}
    }

	/**
	 * 多线程并行导入
	 * @param ds
	 */
	private void batchStreamLoad(DataSource ds) {
		// 以表名进行分组
		Map<String, List<Entity>> tableDataMap = Db.getMapGroupByTableName(beanList);
    	
        // 对每个表并行执行 StreamLoad
    	StreamLoadConfig slc = DbConfig.getStreamLoadConfigByDataSource(ds);
        for (Map.Entry<String, List<Entity>> entry : tableDataMap.entrySet()) {
            List<Entity> dataList = entry.getValue();
            
            HtTaskExecutor.execute(() -> {
            	try {
            		Db.insertBatchWithStream(slc, dataList);
            	} catch (Exception e) {
            		log.error("[oneid]StreamLoad并发导入数据失败：{}", e.getMessage(), e);
            	}
            });
        }
	    
		// 导入完后清空内存
        beanList.clear();
	}
    
    // 添加数据时通知等待的线程
    private void addToMap(List<Entity> entityList) {
        synchronized (LOCK) {
        	beanList.addAll(entityList);
            LOCK.notify();  // 通知等待的线程
        }
    }
    
    /**
     * 根据字段计算分片key
     * @param tenantId 租户ID
     * @param field
     * @return
     */
    private String getKey(long tenantId, String field) {
        int slot = Math.abs(field.hashCode() % HASH_SLOTS);
        return String.format("%s:%s%d", tenantId, KEY, slot);
    }
    
    /**
     * 根据多个字段值，从redis从获取相应的oneid
     * @param tenantId 租户ID
     * @param fields
     * @return
     */
	private <T> List<T> mgetFromRedis(long tenantId, List<String> fields) {
		List<T> oneidList = new ArrayList<>();
		
		for (String field : fields) {
			// 严格按照redis的mget方法来实现，当key为空时，返回一个null对象
			if (StrUtil.isBlank(field)) {
				oneidList.add(null);
			} else {
				String key = getKey(tenantId, field);
				T value = RedisUtil.hget(key, field);
				oneidList.add(value);
			}
		}
		
		return oneidList;
	}

    /**
     * 根据多个字段值，从redis从获取相应的oneid
     * @param tenantId 租户ID
     * @param fields
     * @return
     */
	private void msetToRedis(long tenantId, Map<String, Object> map) {
		for (Map.Entry<String,Object> entry : map.entrySet()) {
			String field = entry.getKey();
			String key = getKey(tenantId, field);
			RedisUtil.hset(key, field, entry.getValue());
		}
	}
	
	public static void main(String[] args) {
		int slot = Math.abs("wmFibqCgAAMg21CPspgol5Ga3tTZhXMw".hashCode() % HASH_SLOTS);
		System.err.println(">>" + slot);
	}
}
